package org.ycy.data.generator.sink;

import com.google.common.collect.Lists;
import lombok.Data;
import lombok.SneakyThrows;
import org.ycy.data.generator.constants.GeneratorConstants;
import org.ycy.data.generator.constants.Symbol;
import org.ycy.data.generator.core.GeneratorFactory;
import org.ycy.data.generator.entity.Column;
import org.ycy.data.generator.entity.TemplateConfig;
import org.ycy.data.generator.enums.StoreType;
import org.ycy.data.generator.generator.IGenerator;

import java.io.File;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @program: data-generator
 * @ClassName Sink
 * @description: 数据落地接口
 * @author: ycy
 * @create: 2023-06-16 22:43
 * @Version 1.0
 **/
@Data

public abstract class Sink {

    protected String headers;
    protected File file;

    protected TemplateConfig templateConfig;


    protected abstract String createHeader();

    @SneakyThrows
    protected void create() {
        StoreType storeType = templateConfig.getStoreType();
        file = new File(templateConfig.getName() + Symbol.POINT + storeType.getSuffix());
        if (file.exists()) {
            file.delete();
        }
        file.createNewFile();
    }

    protected abstract void append(List<List<Object>> data);

    protected void polling() {

        List<Column> columns = templateConfig.getColumns();
        List<List<Object>> data = Lists.newArrayList();
        for (long i = 0; i < templateConfig.getLineSize(); i++) {
            List<Object> list = columns.parallelStream().map(column -> {
                IGenerator iGenerator = GeneratorFactory.buildGenerator(templateConfig.getName(), column);
                return iGenerator.generator();
            }).collect(Collectors.toList());
            data.add(list);
            if (data.size() >= GeneratorConstants.BATCH_SIZE) {
                append(data);
            }
        }
        append(data);
    }

    public void process() {

        create();

        createHeader();

        polling();

        close();
    }

    protected abstract void close();

    public Sink(TemplateConfig templateConfig) {
        this.templateConfig = templateConfig;
    }
}